package hbase.crud.util;

import hbase.crud.entity.Student;
import hbase.crud.entity.StudentInfo;
import hbase.crud.entity.StudentScore;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

public class HbaseClient {
    private Connection conn;
    private static String NS_NAME = "lh1";
    private static String TABLE_NAME = NS_NAME + ":student";

    private static String CF_INFO = "info";
    private static String CF_INFO_STU_ID = "student_id";
    private static String CF_INFO_CLASS = "class";
    private static String CF_SCORE = "score";
    private static String CF_SCORE_UNDERSTANDING = "understanding";
    private static String CF_SCORE_PROGRAMMING = "programming";

    public HbaseClient(String zkServer, int zkPort) {
        try {
            conn = initConnection(zkServer, String.valueOf(zkPort));
        } catch (Throwable t) {
            System.out.println("创建hbase连接异常.");
            t.printStackTrace();
            System.exit(1);
        }
    }

    public void closeClient() {
        if (conn != null) {
            try {
                conn.close();
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    private static Connection initConnection(String zkServer, String zkPort) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkServer);
        conf.set("hbase.zookeeper.property.clientPort", zkPort);
        conf.set("dfs.socket.timeout", "300000");
        return ConnectionFactory.createConnection();
    }

    public void dropStudentTableIfExists() {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        Admin admin = null;
        try {
            admin = conn.getAdmin();

            boolean exists = admin.tableExists(tableName);
            if (exists) {
                System.out.println("旧表存在，进行删除");
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            } else {
                System.out.println("旧表不存在");
            }

            NamespaceDescriptor nd = admin.getNamespaceDescriptor(NS_NAME);
            if (nd != null) {
                System.out.println("namespace存在，进行删除");
                admin.deleteNamespace(NS_NAME);
            }
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }

    public void createStudentTable() {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        Collection<ColumnFamilyDescriptor> families = Arrays.asList(
                ColumnFamilyDescriptorBuilder.newBuilder(CF_INFO.getBytes()).build(),
                ColumnFamilyDescriptorBuilder.newBuilder(CF_SCORE.getBytes()).build()
        );
        NamespaceDescriptor nd = NamespaceDescriptor.create(NS_NAME).build();
        TableDescriptor td = TableDescriptorBuilder.newBuilder(tableName)
                .setColumnFamilies(families)
                .build();

        Admin admin = null;
        try {
            admin = conn.getAdmin();
            admin.createNamespace(nd);
            System.out.println("namespace创建完毕");
            admin.createTable(td);
            System.out.println("table创建完毕");
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (admin != null) {
                try {
                    admin.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }

    public void saveStudent(Student stu) {
        TableName tableName = TableName.valueOf(TABLE_NAME);

        Put put = new Put(Bytes.toBytes(stu.getName()));
        put.addColumn(CF_INFO.getBytes(), CF_INFO_STU_ID.getBytes(), Bytes.toBytes(stu.getInfo().getStudentId()));
        put.addColumn(CF_INFO.getBytes(), CF_INFO_CLASS.getBytes(), Bytes.toBytes(stu.getInfo().getClassId()));
        put.addColumn(CF_SCORE.getBytes(), CF_SCORE_PROGRAMMING.getBytes(), Bytes.toBytes(stu.getScore().getProgramming()));
        put.addColumn(CF_SCORE.getBytes(), CF_SCORE_UNDERSTANDING.getBytes(), Bytes.toBytes(stu.getScore().getUnderstanding()));

        Table table = null;
        try {
            table = conn.getTable(tableName);
            table.put(put);
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }

    public void deleteStudent(String name) {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        Delete del = new Delete(Bytes.toBytes(name));

        Table table = null;
        try {
            table = conn.getTable(tableName);
            table.delete(del);
            System.out.println("删除记录完毕");
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }

    public Student getStudent(String name) {
        TableName tableName = TableName.valueOf(TABLE_NAME);

        Get get = new Get(Bytes.toBytes(name));
        get.addFamily(CF_INFO.getBytes());
        get.addFamily(CF_SCORE.getBytes());

        Table table = null;
        Result rs = null;
        try {
            table = conn.getTable(tableName);
            rs = table.get(get);
            return resultToStudent(rs);
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        return null;
    }

    public List<Student> scanStudent(int maxCount) {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        List<Student> list = new ArrayList<>(maxCount);

        Scan scan = new Scan();
        Table table = null;
        ResultScanner rs = null;
        try {
            table = conn.getTable(tableName);
            rs = table.getScanner(scan);
            Result[] arr = rs.next(maxCount);
            for (Result r : arr) {
                list.add(resultToStudent(r));
            }
        } catch (Throwable t) {
            t.printStackTrace();
        } finally {
            if (rs != null) {
                try {
                    rs.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
            if (table != null) {
                try {
                    table.close();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        return list;
    }

    private Student resultToStudent(Result rs) {
//        System.out.println("rs=" + rs);
//        if (rs != null) {
//            System.out.println(rs.advance());
//            System.out.println(rs.getExists());
//            System.out.println(rs.getRow());
//            System.out.println(rs.isEmpty());
//        }
        if (rs == null || rs.isEmpty()) {
            System.out.println("result数据不存在");
            return null;
        }
        byte[] stuId = rs.getValue(CF_INFO.getBytes(), CF_INFO_STU_ID.getBytes());
        byte[] clsId = rs.getValue(CF_INFO.getBytes(), CF_INFO_CLASS.getBytes());
        byte[] prog = rs.getValue(CF_SCORE.getBytes(), CF_SCORE_PROGRAMMING.getBytes());
        byte[] undst = rs.getValue(CF_SCORE.getBytes(), CF_SCORE_UNDERSTANDING.getBytes());

        StudentInfo info = new StudentInfo();
        info.setStudentId(stuId == null ? null : Bytes.toString(stuId));
        info.setClassId(clsId == null ? null : Bytes.toInt(clsId));

        StudentScore score = new StudentScore();
        score.setUnderstanding(undst == null ? null : Bytes.toInt(undst));
        score.setProgramming(prog == null ? null : Bytes.toInt(prog));

        Student st = new Student();
        st.setName(Bytes.toString(rs.getRow()));
        st.setInfo(info);
        st.setScore(score);
        return st;
    }
}
